package fbcache

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis"
	"github.com/olivere/elastic/v7"
	"github.com/spf13/cast"
	global2 "longmen/server/config/global"
	"longmen/server/internal/rpc/tencentim/imsdk"
	messages2 "longmen/server/internal/rpc/tencentim/messages"
	"longmen/server/internal/rpc/tencentim/messages/sports"
	"reflect"
	sort2 "sort"
	"strings"
	"sync"
	"time"
)

type (
	// TextLive 文字直播数据
	TextLive struct {
		MatchId  int    `json:"match_id"` // 赛程ID
		Time     string `json:"time"`     // 事件时间
		Type     int    `json:"type"`     // 事件类型
		Data     string `json:"data"`     // 事件文本
		Position int    `json:"position"` // 事件发生方，0-中立 1-主队 2-客队
		Main     int    `json:"main"`     // 是否重要事件
	}
	EsTextLive struct {
		MatchId    int    `json:"match_id"`    // 赛程ID
		Time       string `json:"time"`        // 事件时间
		Type       int    `json:"type"`        // 事件类型
		Data       string `json:"data"`        // 事件文本
		Position   int    `json:"position"`    // 事件发生方，0-中立 1-主队 2-客队
		Main       int    `json:"main"`        // 是否重要事件
		CreateTime int64  `json:"create_time"` // 写入数据时间
	}
	TextLiveCache struct {
		rds redis.UniversalClient
		es  *elastic.Client
	}
)

var textLiveCache *TextLiveCache
var textLiveCacheOnce sync.Once

func NewTextLiveCache() *TextLiveCache {
	textLiveCacheOnce.Do(func() {
		textLiveCache = &TextLiveCache{
			rds: global2.GetRds(),
			es:  global2.GetEs(),
		}
	})
	return textLiveCache
}

// 生成足球文字直播key
func (t *TextLiveCache) genTextLiveKey(matchId int) string {
	return fmt.Sprintf("match_live:football_text_live:%d", matchId)
}

func (t *TextLiveCache) PublishTextLive(ive []TextLive) error {
	dataMap := make(map[int][]TextLive)
	for i, record := range ive {
		temp := record
		// 此处为了兼容老版本，因为老版本ronnie写的是顺序ID
		temp.MatchId = i + 1
		dataMap[record.MatchId] = append(dataMap[record.MatchId], temp)
	}
	pp := t.rds.Pipeline()
	for id, textLive := range dataMap {
		message := sports.TextLiveMessage{
			SentMessageStruct: messages2.SentMessageStruct{Type: messages2.RoomTextLiveMessageType},
			TextLive:          textLive,
		}
		marshal, _ := json.Marshal(message)
		err := pp.Set(t.genTextLiveKey(id), string(marshal), time.Hour*24).Err()
		if err != nil {
			global2.Log.Error("err: ", err)
		}
		imsdk.GetSdk().BroadcastInterface(id, message, 0)
	}
	_, err := pp.Exec()
	if err != nil {
		global2.Log.Errorf("Pipeline run error: %v", err)
	}
	t.writeToes(ive)
	return nil
}

// GetTextLiveCache 获取缓存
func (t *TextLiveCache) GetTextLiveCache(matchId int) string {
	result, err := t.rds.Get(t.genTextLiveKey(matchId)).Result()
	if err != nil {
		return ""
	}
	return result
}

// 文字直播KEY
func (t *TextLiveCache) getEsIndexName() string {
	return fmt.Sprintf("football_text_live")
}
func (t *TextLiveCache) InitEsIndex() {
	t.es.Index().Index(t.getEsIndexName())
}

func (t *TextLiveCache) writeToes(sendData []TextLive) {
	esRecords := make([]EsTextLive, 0)
	for _, item := range sendData {
		esRecords = append(esRecords, EsTextLive{
			MatchId:  item.MatchId,
			Time:     item.Time,
			Type:     item.Type,
			Data:     item.Data,
			Position: item.Position,
			Main:     item.Main,
			//CreateTime: timehelper.Now().UnixNano(), // 没用，会导致乱序
		})
	}

	bulks := make([]*elastic.BulkIndexRequest, 0)
	total := 3000 // 3000条批量写一次
	writeCount := 0
	for _, datum := range esRecords {
		// 赛程ID+timehelper
		id := fmt.Sprintf("%d-%s", datum.MatchId, datum.Time)
		bulks = append(bulks, elastic.NewBulkIndexRequest().Index(t.getEsIndexName()).Doc(datum).Id(id))
		writeCount++
		if writeCount == total {
			t.bulkToEs(bulks)
			bulks = make([]*elastic.BulkIndexRequest, 0)
		}
	}

	if len(bulks) > 0 {
		t.bulkToEs(bulks)
	}

}

func (t *TextLiveCache) bulkToEs(records []*elastic.BulkIndexRequest) {
	bulk := t.es.Bulk()
	for i := range records {
		bulk.Add(records[i])
	}
	res, err := bulk.Do(context.Background())
	if err != nil {
		global2.Log.Errorf("exec bulk to es : %v", err)
		return
	}
	for _, item := range res.Failed() {
		global2.Log.Errorf("write to es err: %s", item.Error.Reason)
	}
}

func (t *TextLiveCache) SearchByEs(matchId int, limit int) []messages2.TextLiveStruct {
	sort := elastic.NewFieldSort("create_time").Asc()
	searcher := t.es.Search(t.getEsIndexName()).Size(limit)
	query := elastic.NewBoolQuery()
	query.Filter(elastic.NewMatchQuery("match_id", matchId))
	res, err := searcher.PostFilter(query).SortBy(sort).Do(context.Background())
	if err != nil {
		global2.Log.Errorf("search error: %v", err)
		return nil
	}
	var item messages2.TextLiveStruct
	records := make([]messages2.TextLiveStruct, 0)
	for _, v := range res.Each(reflect.TypeOf(item)) {
		textData := v.(messages2.TextLiveStruct)
		timeStr := strings.Split(textData.Time, "'")
		if len(timeStr) > 0 {
			textData.Id = cast.ToInt(timeStr[0])
		}
		records = append(records, textData)
	}
	sort2.Slice(records, func(x, y int) bool {
		if records[x].Id > records[y].Id {
			return true
		}
		return false
	})
	//global.Log.Debugf("足球文字直播log:%+v", records)
	return records
}
